Skip to content

0.1 Python 核心基础

学习目标 对于有一定 Python 基础的朋友,本章可以跳过,从第一章开始你的智能体搭建之旅。

本章专为 LangGraph 和 AI Agent 开发设计,通过实战代码快速掌握 Python 核心语法。我们采用"先看代码,后学原理"的方式,每个概念都关联实际的 Agent 开发场景。


本章内容

  • ✅ 掌握 Python 变量与数据类型
  • ✅ 理解数据结构(List、Dict、Tuple、Set)在 Agent 中的应用
  • ✅ 学会控制流(条件判断、循环)
  • ✅ 掌握函数定义与 Lambda 表达式
  • ✅ 理解类型提示(为 LangGraph 做准备)

📚 术语表

术语名称LangGraph 定义和解读Python 定义和说明重要程度
StateAgent的状态字典,存储消息历史、意图、上下文等信息。LangGraph中每个节点接收state并返回更新后的state字典(dict)类型,键值对集合,可变数据结构⭐⭐⭐⭐⭐
TypedDict定义严格的State结构,类似TypeScript的Interface。确保状态字段类型安全,便于IDE智能提示typing模块提供的类型,为字典添加类型注解,支持静态类型检查⭐⭐⭐⭐⭐
Lambda用于条件边(Conditional Edge)的简洁路由函数,如lambda s: "next_node",快速定义节点转换逻辑匿名函数,格式lambda 参数: 表达式,适合简单单行逻辑⭐⭐⭐⭐
List存储消息历史(messages)、工具调用记录等有序数据。Agent核心数据结构有序可变序列,支持索引、切片、append等操作⭐⭐⭐⭐⭐
DictAgent状态的基础结构,键值对存储配置、上下文、元数据无序可变映射,通过键访问值,支持get/items/keys等方法⭐⭐⭐⭐⭐
f-string构建LLM提示词、格式化日志输出的现代方式,如f"用户{name}的意图是{intent}"格式化字符串字面量,f"...{变量}..."支持表达式求值⭐⭐⭐⭐
Type HintsLangGraph要求的类型注解,定义节点函数签名def node(state: AgentState) -> AgentStatePython 3.5+特性,为变量和函数添加类型信息,增强代码可读性⭐⭐⭐⭐⭐
Literal限定字段只能是特定值,如role: Literal["user", "assistant"]避免拼写错误typing.Literal类型,定义字面量类型,编译时检查值合法性⭐⭐⭐⭐
Optional表示函数可能返回None,如Optional[str]等价于Union[str, None]typing.Optional类型,明确告知调用方需处理空值情况⭐⭐⭐⭐
List Comprehension简洁处理消息过滤、数据转换,如[msg for msg in messages if msg["role"]=="user"]列表推导式,一行代码完成循环+过滤+映射,性能优于传统循环⭐⭐⭐⭐

1. 环境准备:开始你的第一行代码

推荐开发环境

选项 1:在线运行(最简单)

  • 本网站内置 Python 运行环境,无需安装
  • 需要运行 LangGraph 代码时,请先配置 API Key

选项 2:本地 Jupyter Lab(推荐学习。可搭配 Anaconda 使用)

bash
# 安装 Jupyter Lab
pip install jupyterlab

# 启动
jupyter lab

选项 3:Cursor / VS Code(推荐开发)

bash
# 安装扩展:
# 1. Python (by Microsoft)
# 2. Jupyter (by Microsoft)

Hello, AI World!

运行你的第一个 Python 程序:

python
# 验证环境
import sys
print(f"Python 版本: {sys.version}")
print(f"🤖 你好,AI Agent 世界!")

预期结果:

Python 版本: 3.13.0 (main, Oct  7 2024, 05:02:14) [Clang 15.0.0]
🤖 你好,AI Agent 世界!

2. 变量与数据类型

30秒快速上手:最简变量示例

如果你是完全的 Python 小白,先看这个最简单的例子:

python
# 最简单的变量例子 - 就像给东西贴标签
name = "小明"           # 文字用引号括起来
age = 18               # 数字直接写
height = 1.75          # 小数点的数字
is_student = True      # True 或 False(注意大写)

# 打印出来看看
print(name)            # 输出:小明
print(age)             # 输出:18
print(height)          # 输出:1.75
print(is_student)      # 输出:True

# 组合打印
print(name, "今年", age, "岁")  # 输出:小明 今年 18 岁

💡 新手提示

  • 变量名就像给数据起个名字,方便后面使用
  • = 是赋值,把右边的值给左边的变量
  • print() 用来显示结果

2.1 变量:存储 Agent 状态

在 AI Agent 中,变量用于存储用户消息、系统状态、对话历史等关键信息。

python
# 代码示例:Agent 基本状态
user_message = "帮我查一下今天的天气"
user_id = "user_12345"
message_count = 3
is_authenticated = True
confidence_score = 0.85

print(f"用户消息: {user_message}")
print(f"用户ID: {user_id}")
print(f"消息数量: {message_count}")
print(f"已认证: {is_authenticated}")
print(f"置信度: {confidence_score}")

# 查看类型
print(f"\n数据类型:")
print(f"  user_message: {type(user_message)}")
print(f"  message_count: {type(message_count)}")
print(f"  confidence_score: {type(confidence_score)}")
print(f"  is_authenticated: {type(is_authenticated)}")

预期结果:

用户消息: 帮我查一下今天的天气
用户ID: user_12345
消息数量: 3
已认证: True
置信度: 0.85

数据类型:
  user_message: <class 'str'>
  message_count: <class 'int'>
  confidence_score: <class 'float'>
  is_authenticated: <class 'bool'>

核心数据类型

类型说明AI 应用场景
str字符串(文本)用户消息、AI 回复、提示词
int整数消息计数、步骤编号、重试次数
float浮点数置信度分数、温度参数、相似度
bool布尔值条件判断、状态标志、工具调用成功与否
None空值可选参数、未初始化的状态

2.2 字符串操作:处理文本

python
# 代码示例:字符串在 Agent 中的常见操作
user_input = "   帮我预订明天去北京的机票   "

# 清理输入
cleaned = user_input.strip()  # 去除首尾空格
print(f"清理后: '{cleaned}'")

# 大小写转换
lowercase = cleaned.lower()
uppercase = cleaned.upper()
print(f"小写: {lowercase}")
print(f"大写: {uppercase}")

# 字符串拼接
prompt = f"用户请求:{cleaned}\n请分析意图并给出建议。"
print(f"\n构建的提示词:\n{prompt}")

# 字符串分割
keywords = "天气,航班,酒店,餐厅"
keyword_list = keywords.split(",")
print(f"\n关键词列表: {keyword_list}")

# 检查子串
if "预订" in cleaned:
    print("\n✅ 检测到预订意图")

# 字符串替换
sanitized = cleaned.replace("明天", "2024-10-28")
print(f"日期规范化: {sanitized}")

预期结果:

清理后: '帮我预订明天去北京的机票'
小写: 帮我预订明天去北京的机票
大写: 帮我预订明天去北京的机票

构建的提示词:
用户请求:帮我预订明天去北京的机票
请分析意图并给出建议。

关键词列表: ['天气', '航班', '酒店', '餐厅']

✅ 检测到预订意图
日期规范化: 帮我预订2024-10-28去北京的机票

2.3 f-string:现代字符串格式化

python
# 代码示例:构建 AI 提示词
user_name = "Alice"
city = "北京"
temperature = 0.7
max_tokens = 1000

# f-string 基础
greeting = f"你好,{user_name}!"
print(greeting)

# 表达式求值
tokens_used = 850
remaining = f"已使用 {tokens_used}/{max_tokens} tokens ({tokens_used/max_tokens:.1%})"
print(remaining)

# 多行 f-string(构建复杂提示词)
system_prompt = f"""你是一个智能助手。
用户信息:
- 姓名:{user_name}
- 位置:{city}

参数设置:
- Temperature: {temperature}
- Max Tokens: {max_tokens}

请根据用户需求提供帮助。"""

print(f"\n{system_prompt}")

# 格式化数字
pi = 3.14159265359
print(f"\nπ ≈ {pi:.2f}")  # 保留2位小数
print(f"π ≈ {pi:.5f}")  # 保留5位小数

预期结果:

你好,Alice!
已使用 850/1000 tokens (85.0%)

你是一个智能助手。
用户信息:
- 姓名:Alice
- 位置:北京

参数设置:
- Temperature: 0.7
- Max Tokens: 1000

请根据用户需求提供帮助。

π ≈ 3.14
π ≈ 3.14159

3. 数据结构:组织 Agent 数据

🚀 30秒快速上手:最简列表和字典

如果你是 Python 小白,先看这两个最常用的数据结构:

python
# 列表 - 就像一个有序的购物清单
fruits = ["苹果", "香蕉", "橙子"]
print(fruits[0])        # 输出:苹果(第一个,从0开始数)
print(fruits[-1])       # 输出:橙子(最后一个)
fruits.append("葡萄")   # 添加新水果
print(fruits)           # 输出:['苹果', '香蕉', '橙子', '葡萄']

# 字典 - 就像通讯录,有名字就能找到对应信息
student = {"姓名": "小明", "年龄": 18, "成绩": 95}
print(student["姓名"])   # 输出:小明
student["年龄"] = 19     # 修改年龄
print(student)          # 输出:{'姓名': '小明', '年龄': 19, '成绩': 95}

💡 新手提示

  • 列表用方括号 [],按顺序存放多个东西
  • 字典用花括号 {},用"名字:值"的方式存放,方便查找

3.1 列表(List):对话历史管理

列表是 AI Agent 中最常用的数据结构,用于存储消息序列、工具调用记录等。

python
# 代码示例:管理对话历史
messages = []

# 添加用户消息
messages.append({"role": "user", "content": "你好"})
messages.append({"role": "assistant", "content": "你好!有什么可以帮你的吗?"})
messages.append({"role": "user", "content": "今天天气怎么样?"})

# 访问消息
print("=== 对话历史 ===")
print(f"总消息数: {len(messages)}")
print(f"第一条消息: {messages[0]}")
print(f"最后一条消息: {messages[-1]}")
print(f"最近两条消息: {messages[-2:]}")

# 遍历消息
print("\n=== 完整对话 ===")
for i, msg in enumerate(messages, 1):
    role_icon = "👤" if msg["role"] == "user" else "🤖"
    print(f"{i}. {role_icon} {msg['content']}")

# 列表操作
messages.insert(0, {"role": "system", "content": "你是一个友好的助手"})
print(f"\n添加系统消息后,总消息数: {len(messages)}")

# 删除消息
removed = messages.pop(1)  # 删除索引1的消息
print(f"删除的消息: {removed}")

# 列表推导式(过滤用户消息)
user_messages = [msg for msg in messages if msg["role"] == "user"]
print(f"\n用户消息: {user_messages}")

预期结果:

=== 对话历史 ===
总消息数: 3
第一条消息: {'role': 'user', 'content': '你好'}
最后一条消息: {'role': 'user', 'content': '今天天气怎么样?'}
最近两条消息: [{'role': 'assistant', 'content': '你好!有什么可以帮你的吗?'}, {'role': 'user', 'content': '今天天气怎么样?'}]

=== 完整对话 ===
1. 👤 你好
2. 🤖 你好!有什么可以帮你的吗?
3. 👤 今天天气怎么样?

添加系统消息后,总消息数: 4
删除的消息: {'role': 'user', 'content': '你好'}

用户消息: [{'role': 'user', 'content': '今天天气怎么样?'}]

3.2 字典(Dict):Agent 状态管理

字典是 LangGraph 的核心数据结构,用于存储 Agent 的状态信息。

python
# 代码示例:LangGraph 状态对象
agent_state = {
    "user_id": "user_12345",
    "session_id": "sess_abc",
    "messages": [
        {"role": "user", "content": "帮我查天气"},
        {"role": "assistant", "content": "好的,请问是哪个城市?"}
    ],
    "context": {
        "intent": "weather_query",
        "location": None,
        "confidence": 0.92
    },
    "tool_calls": [],
    "metadata": {
        "start_time": "2024-10-27 10:00:00",
        "turn_count": 1
    }
}

# 访问数据
print(f"用户ID: {agent_state['user_id']}")
print(f"意图: {agent_state['context']['intent']}")
print(f"置信度: {agent_state['context']['confidence']:.1%}")

# 更新状态
agent_state['context']['location'] = "北京"
agent_state['metadata']['turn_count'] += 1

print(f"\n更新后的位置: {agent_state['context']['location']}")
print(f"对话轮次: {agent_state['metadata']['turn_count']}")

# 添加新字段
agent_state['context']['weather_data'] = {
    "temperature": 22,
    "condition": "晴朗"
}

# 获取所有键
print(f"\n状态顶层字段: {list(agent_state.keys())}")
print(f"上下文字段: {list(agent_state['context'].keys())}")

# 安全访问(使用 get)
api_key = agent_state.get('api_key', 'NOT_SET')
print(f"\nAPI Key: {api_key}")

# 检查键是否存在
if 'tool_calls' in agent_state:
    print("✅ 状态包含 tool_calls 字段")

# 字典遍历
print("\n=== 元数据详情 ===")
for key, value in agent_state['metadata'].items():
    print(f"  {key}: {value}")

预期结果:

用户ID: user_12345
意图: weather_query
置信度: 92.0%

更新后的位置: 北京
对话轮次: 2

状态顶层字段: ['user_id', 'session_id', 'messages', 'context', 'tool_calls', 'metadata']
上下文字段: ['intent', 'location', 'confidence', 'weather_data']

API Key: NOT_SET
✅ 状态包含 tool_calls 字段

=== 元数据详情 ===
  start_time: 2024-10-27 10:00:00
  turn_count: 2

3.3 元组(Tuple):不可变数据

python
# 代码示例:消息格式、配置常量
# 元组用于固定格式的数据
message_format = ("role", "content", "timestamp")
print(f"消息格式: {message_format}")

# 解包元组
role, content, timestamp = message_format
print(f"字段1: {role}, 字段2: {content}, 字段3: {timestamp}")

# 多返回值(实际是元组)
def parse_message(text):
    """解析消息,返回意图和实体"""
    intent = "weather_query"
    entity = "北京"
    confidence = 0.95
    return intent, entity, confidence  # 返回元组

intent, entity, conf = parse_message("北京天气")
print(f"\n解析结果: 意图={intent}, 实体={entity}, 置信度={conf}")

# 元组不可修改
try:
    message_format[0] = "new_role"
except TypeError as e:
    print(f"\n❌ 元组不可修改: {e}")

# 元组作为字典键(列表不行)
cache = {
    ("user_123", "weather"): "晴天 22°C",
    ("user_456", "weather"): "多云 18°C"
}
print(f"\n缓存查询: {cache[('user_123', 'weather')]}")

预期结果:

消息格式: ('role', 'content', 'timestamp')
字段1: role, 字段2: content, 字段3: timestamp

解析结果: 意图=weather_query, 实体=北京, 置信度=0.95

❌ 元组不可修改: 'tuple' object does not support item assignment

缓存查询: 晴天 22°C

3.4 集合(Set):去重与快速查找

python
# 代码示例:管理用户兴趣标签
user_interests = {"科技", "旅游", "美食", "科技", "运动"}  # 自动去重
print(f"用户兴趣(已去重): {user_interests}")

# 添加兴趣
user_interests.add("阅读")
user_interests.add("科技")  # 重复添加无效
print(f"添加后: {user_interests}")

# 集合运算
available_topics = {"科技", "美食", "娱乐", "财经"}

# 交集:用户感兴趣且可用的话题
common = user_interests & available_topics
print(f"\n推荐话题(交集): {common}")

# 并集:所有相关话题
all_topics = user_interests | available_topics
print(f"所有话题(并集): {all_topics}")

# 差集:用户兴趣但不可用的
missing = user_interests - available_topics
print(f"缺失话题(差集): {missing}")

# 快速检查
if "科技" in user_interests:
    print("\n✅ 用户对科技感兴趣")

# 转换:列表 → 集合(去重)
message_keywords = ["天气", "查询", "天气", "北京", "查询"]
unique_keywords = set(message_keywords)
print(f"\n关键词去重: {list(unique_keywords)}")

预期结果:

用户兴趣(已去重): {'运动', '旅游', '科技', '美食'}
添加后: {'运动', '旅游', '阅读', '科技', '美食'}

推荐话题(交集): {'科技', '美食'}
所有话题(并集): {'娱乐', '运动', '旅游', '阅读', '财经', '科技', '美食'}
缺失话题(差集): {'运动', '旅游', '阅读'}

✅ 用户对科技感兴趣

关键词去重: ['查询', '天气', '北京']

📊 数据结构选择指南

数据结构有序可变去重使用场景
List对话历史、消息队列、工具调用记录
DictKey去重Agent 状态、配置参数、API 响应
Tuple固定格式、函数返回、常量定义
Set标签去重、权限检查、集合运算

4. 控制流:Agent 决策逻辑

🚀 30秒快速上手:最简 if 和 for

如果你是 Python 小白,先看这两个最基础的控制语句:

python
# if - 条件判断,就像"如果...就..."
score = 85
if score >= 90:
    print("优秀")
elif score >= 60:
    print("及格")      # 这个会被执行
else:
    print("不及格")

# for - 循环,重复做事情
fruits = ["苹果", "香蕉", "橙子"]
for fruit in fruits:
    print("我喜欢", fruit)
# 输出:
# 我喜欢 苹果
# 我喜欢 香蕉
# 我喜欢 橙子

💡 新手提示

  • if 后面的代码要缩进(按 Tab 键或 4 个空格)
  • for 会依次取出列表里的每一项

4.1 条件判断:路由与决策

在 LangGraph 中,条件判断用于实现条件边(Conditional Edge),决定 Agent 的下一步行为。

python
# 代码示例:根据用户意图路由
def route_by_intent(user_message):
    """根据用户消息路由到不同节点"""
    # 【设计思想】先统一转小写,避免大小写不匹配问题
    # 这是文本预处理的标准实践
    message_lower = user_message.lower()

    # 【核心逻辑】基于关键词匹配的意图识别
    # 类似决策树:从最明确的意图开始判断
    if "天气" in message_lower:
        # 高置信度(0.9):天气是单一明确的意图
        intent = "weather_query"
        next_node = "weather_tool"
        confidence = 0.9
    elif "预订" in message_lower or "订票" in message_lower:
        # 稍低置信度(0.85):预订可能需要更多澄清
        intent = "booking"
        next_node = "booking_tool"
        confidence = 0.85
    elif "计算" in message_lower or "=" in message_lower:
        # 最高置信度(0.95):计算特征非常明显
        intent = "calculator"
        next_node = "calculator_tool"
        confidence = 0.95
    else:
        # 【回退策略】未匹配时交给通用LLM处理
        # 低置信度(0.5)提示需要人工审核
        intent = "general_chat"
        next_node = "llm_response"
        confidence = 0.5

    # 【返回结构】包含决策结果和置信度,便于后续审计和调试
    return {
        "intent": intent,
        "next_node": next_node,
        "confidence": confidence
    }

# 测试路由
test_messages = [
    "今天北京天气怎么样?",
    "帮我预订明天去上海的机票",
    "计算 123 + 456",
    "你好,介绍一下你自己"
]

print("=== Intent Routing ===")
for msg in test_messages:
    result = route_by_intent(msg)
    print(f"\n消息: {msg}")
    print(f"  → 意图: {result['intent']}")
    print(f"  → 下一节点: {result['next_node']}")
    print(f"  → 置信度: {result['confidence']:.0%}")

预期结果:

=== Intent Routing ===

消息: 今天北京天气怎么样?
  → 意图: weather_query
  → 下一节点: weather_tool
  → 置信度: 90%

消息: 帮我预订明天去上海的机票
  → 意图: booking
  → 下一节点: booking_tool
  → 置信度: 85%

消息: 计算 123 + 456
  → 意图: calculator
  → 下一节点: calculator_tool
  → 置信度: 95%

消息: 你好,介绍一下你自己
  → 意图: general_chat
  → 下一节点: llm_response
  → 置信度: 50%

4.2 三元运算符:简洁的条件表达式

python
# 代码示例:快速条件赋值
def evaluate_response_quality(confidence_score):
    """评估响应质量"""
    # 【传统写法】标准 if-else 分支
    if confidence_score > 0.8:
        quality = "高质量"
    else:
        quality = "需要确认"

    # 【三元运算符】一行代码实现条件赋值
    # 格式: 值1 if 条件 else 值2
    # 适用于简单的二选一场景,代码更简洁
    quality_concise = "高质量" if confidence_score > 0.8 else "需要确认"

    # 【嵌套三元运算符】实现多级分类
    # 逻辑链: >0.9→优秀, >0.7→良好, 其他→一般
    # 注意:嵌套超过2层会降低可读性,应改用if-elif
    level = "优秀" if confidence_score > 0.9 else "良好" if confidence_score > 0.7 else "一般"

    return {
        "quality": quality_concise,
        "level": level,
        "should_confirm": confidence_score < 0.8  # 布尔表达式直接作为值
    }

# 测试
scores = [0.95, 0.75, 0.5]
for score in scores:
    result = evaluate_response_quality(score)
    print(f"置信度 {score}: {result}")

预期结果:

置信度 0.95: {'quality': '高质量', 'level': '优秀', 'should_confirm': False}
置信度 0.75: {'quality': '需要确认', 'level': '良好', 'should_confirm': True}
置信度 0.5: {'quality': '需要确认', 'level': '一般', 'should_confirm': True}

4.3 for 循环:批量处理

python
# 代码示例:批量处理消息
messages = [
    {"id": 1, "content": "天气查询", "processed": False},
    {"id": 2, "content": "预订机票", "processed": False},
    {"id": 3, "content": "闲聊", "processed": False}
]

print("=== 处理消息队列 ===")
# 【enumerate技巧】同时获取索引i和元素msg
# enumerate(list)返回(索引,元素)元组,避免手动维护计数器
for i, msg in enumerate(messages):
    print(f"\n处理消息 #{msg['id']}: {msg['content']}")

    # 【原地修改】直接修改字典,无需重新赋值
    # Python中字典是可变对象,循环中的修改会影响原列表
    msg['processed'] = True
    msg['timestamp'] = f"2024-10-27 10:0{i}:00"

    print(f"  ✅ 已处理,时间戳: {msg['timestamp']}")

# 【列表推导式】一行代码完成过滤+提取
# 等价于: for循环 + if判断 + append操作
# 比传统循环更简洁,性能也略优
processed_ids = [msg['id'] for msg in messages if msg['processed']]
print(f"\n已处理的消息ID: {processed_ids}")

# 【字典推导式】快速构建映射关系
# 格式: {key表达式: value表达式 for item in iterable}
id_to_content = {msg['id']: msg['content'] for msg in messages}
print(f"ID映射: {id_to_content}")

# 【range循环】生成数字序列
# range(1, 4)生成[1,2,3],注意右边界不包含
print("\n=== 重试机制 ===")
max_retries = 3
for attempt in range(1, max_retries + 1):
    print(f"尝试 #{attempt}/{max_retries}")
    success = (attempt == 2)  # 模拟第2次成功
    if success:
        print("  ✅ 成功!")
        break  # 成功后立即退出循环,避免浪费重试次数
    print("  ❌ 失败,重试中...")

预期结果:

=== 处理消息队列 ===

处理消息 #1: 天气查询
  ✅ 已处理,时间戳: 2024-10-27 10:00:00

处理消息 #2: 预订机票
  ✅ 已处理,时间戳: 2024-10-27 10:01:00

处理消息 #3: 闲聊
  ✅ 已处理,时间戳: 2024-10-27 10:02:00

已处理的消息ID: [1, 2, 3]
ID映射: {1: '天气查询', 2: '预订机票', 3: '闲聊'}

=== 重试机制 ===
尝试 #1/3
  ❌ 失败,重试中...
尝试 #2/3
  ✅ 成功!

4.4 while 循环:条件驱动

python
# 代码示例:API 重试逻辑
import time

def call_api_with_retry(max_retries=3, delay=1):
    """模拟带重试的 API 调用"""
    attempt = 0

    # 【while循环】条件驱动,适合未知循环次数的场景
    # 这里用于重试:只要未达上限且未成功,就继续尝试
    while attempt < max_retries:
        attempt += 1  # 先递增计数器
        print(f"API 调用尝试 #{attempt}...")

        # 【模拟调用】实际开发中这里会是真实的API请求
        success = (attempt == 2)

        if success:
            # 【成功返回】立即退出函数,无需继续循环
            print("  ✅ API 调用成功!")
            return {"status": "success", "data": "天气数据"}
        else:
            print(f"  ❌ 失败,{delay}秒后重试...")
            # 【指数退避】实际应用中delay应递增(如1,2,4,8秒)
            # 这里简化为固定延迟,避免快速耗尽API配额
            if attempt < max_retries:
                time.sleep(delay)

    # 【兜底处理】循环正常结束说明所有重试都失败
    # 返回错误而非抛出异常,让调用方决定如何处理
    print("\n⚠️ 达到最大重试次数,调用失败")
    return {"status": "failed", "error": "Max retries exceeded"}

# 执行
result = call_api_with_retry(max_retries=3, delay=0.5)
print(f"\n最终结果: {result}")

# 【break和continue】控制循环流程
print("\n=== 处理消息(跳过无效消息)===")
messages = ["valid_msg_1", "", "valid_msg_2", None, "valid_msg_3"]
processed_count = 0

for msg in messages:
    # 【continue】跳过本次循环,直接进入下一次迭代
    # if not msg 等价于 if msg == "" or msg == None
    if not msg:  # 跳过空消息
        print(f"⏭️  跳过无效消息")
        continue  # 不执行后续代码,直接下一个msg

    print(f"✅ 处理: {msg}")
    processed_count += 1

    # 【break】彻底退出循环,不再处理后续元素
    # 常用于找到目标值、达到处理上限等场景
    if processed_count >= 2:  # 只处理2条
        print("⏸️  达到处理上限,停止")
        break  # 退出for循环

print(f"\n总共处理: {processed_count} 条消息")

预期结果:

API 调用尝试 #1...
  ❌ 失败,0.5秒后重试...
API 调用尝试 #2...
  ✅ API 调用成功!

最终结果: {'status': 'success', 'data': '天气数据'}

=== 处理消息(跳过无效消息)===
✅ 处理: valid_msg_1
⏭️  跳过无效消息
✅ 处理: valid_msg_2
⏸️  达到处理上限,停止

总共处理: 2 条消息

5. 函数:代码复用的艺术

🚀 30秒快速上手:最简函数

如果你是 Python 小白,先看这个最简单的函数例子:

python
# 定义函数 - 就像给一段代码起个名字,方便重复使用
def greet(name):
    print("你好,", name)
    return "欢迎"

# 调用函数
message = greet("小明")    # 输出:你好, 小明
print(message)            # 输出:欢迎

# 带计算的函数
def add(a, b):
    result = a + b
    return result

total = add(5, 3)         # 调用函数
print(total)              # 输出:8

💡 新手提示

  • def 用来定义函数,后面跟函数名和括号
  • 括号里是参数(输入),return返回值(输出)
  • 定义后要调用才会执行

5.1 函数基础:LangGraph 节点函数

在 LangGraph 中,每个节点都是一个函数,接收状态并返回更新后的状态。

python
# 代码示例:定义 Agent 节点函数
def process_user_message(state):
    """处理用户消息的节点函数

    【LangGraph核心概念】
    节点函数是图中的计算单元,接收状态、处理逻辑、返回更新后的状态
    这种函数签名(state → state)是LangGraph的标准模式

    Args:
        state: Agent 状态字典

    Returns:
        更新后的状态
    """
    # 【状态读取】从消息列表获取最新用户输入
    # [-1]索引获取列表最后一个元素,这是Python的惯用法
    user_message = state["messages"][-1]["content"]

    # 【意图识别】简单的关键词匹配
    # 实际生产中应使用LLM或分类模型
    intent = "unknown"
    if "天气" in user_message:
        intent = "weather"
    elif "预订" in user_message:
        intent = "booking"

    # 【状态更新】直接修改state字典
    # LangGraph会自动合并这些更新到全局状态
    state["intent"] = intent
    state["processed"] = True

    print(f"📥 处理消息: {user_message}")
    print(f"🎯 识别意图: {intent}")

    # 【必须返回state】这是LangGraph节点函数的契约
    return state

# 测试节点函数
initial_state = {
    "messages": [
        {"role": "user", "content": "今天天气怎么样?"}
    ],
    "intent": None,
    "processed": False
}

updated_state = process_user_message(initial_state)
print(f"\n更新后的状态: {updated_state}")

预期结果:

📥 处理消息: 今天天气怎么样?
🎯 识别意图: weather

更新后的状态: {'messages': [{'role': 'user', 'content': '今天天气怎么样?'}], 'intent': 'weather', 'processed': True}

5.2 参数与返回值

python
# 代码示例:灵活的函数参数
def create_prompt(
    system_role="helpful assistant",  # 【默认参数】未传入时使用此值
    user_message=None,                # 【可选参数】允许为None,后续校验
    temperature=0.7,                  # 【模型参数】OpenAI推荐0-2范围
    max_tokens=1000,
    **kwargs                          # 【可变关键字参数】捕获所有额外参数
):
    """构建 LLM 提示词

    【参数设计原则】
    - 必需参数放前面(user_message)
    - 常用参数有默认值(system_role, temperature)
    - **kwargs处理未知参数,增强扩展性

    Args:
        system_role: 系统角色
        user_message: 用户消息
        temperature: 温度参数
        max_tokens: 最大 token 数
        **kwargs: 其他参数(如top_p, model等)

    Returns:
        dict: 提示词配置
    """
    # 【参数校验】早失败(fail-fast)原则
    # 在函数开始就检查必需参数,避免后续计算浪费
    if user_message is None:
        raise ValueError("必须提供 user_message")

    # 【配置构建】标准化API调用格式
    config = {
        "system": system_role,
        "user": user_message,
        "parameters": {
            "temperature": temperature,
            "max_tokens": max_tokens
        }
    }

    # 【额外参数处理】灵活支持新功能而无需修改函数签名
    # 例如:未来OpenAI新增参数,调用方可直接传入
    if kwargs:
        config["extra"] = kwargs

    return config

# 使用函数
prompt1 = create_prompt(user_message="你好")
print("基础调用:")
print(prompt1)

prompt2 = create_prompt(
    system_role="专业的天气助手",
    user_message="北京天气?",
    temperature=0.5,
    top_p=0.9,  # 额外参数会被**kwargs捕获
    model="gpt-5"
)
print("\n完整调用:")
print(prompt2)

# 【返回多个值】Python特性:用元组打包多个返回值
def analyze_sentiment(text):
    """分析文本情感"""
    # 【简化分析】实际应使用Transformers情感分析模型
    score = 0.8 if "好" in text else 0.3
    label = "积极" if score > 0.5 else "消极"
    confidence = score

    # Python会自动将多个值打包成元组
    return score, label, confidence  # 返回元组

# 【元组解包】一次性接收多个返回值
# 这比返回字典再提取更简洁
score, label, conf = analyze_sentiment("今天天气真好!")
print(f"\n情感分析: 得分={score}, 标签={label}, 置信度={conf}")

预期结果:

基础调用:
{'system': 'helpful assistant', 'user': '你好', 'parameters': {'temperature': 0.7, 'max_tokens': 1000}}

完整调用:
{'system': '专业的天气助手', 'user': '北京天气?', 'parameters': {'temperature': 0.5, 'max_tokens': 1000}, 'extra': {'top_p': 0.9, 'model': 'gpt-4'}}

情感分析: 得分=0.8, 标签=积极, 置信度=0.8

5.3 Lambda 函数:条件边的利器

python
# 代码示例:Lambda 在 LangGraph 中的应用

# 【传统函数】适合复杂逻辑
def should_continue(state):
    return state.get("continue", True)

# 【Lambda等价写法】适合简单单行逻辑
# 格式: lambda 参数: 表达式
# 优点:简洁,常用于临时函数 缺点:调试困难,不支持多行
should_continue_lambda = lambda state: state.get("continue", True)

# 【Lambda用于排序】最常见的应用场景
messages = [
    {"id": 3, "content": "third", "priority": 2},
    {"id": 1, "content": "first", "priority": 1},
    {"id": 2, "content": "second", "priority": 3}
]

# key参数接受函数,定义排序依据
# lambda x: x["priority"] 表示"按priority字段排序"
sorted_by_priority = sorted(messages, key=lambda x: x["priority"])
print("按优先级排序:")
for msg in sorted_by_priority:
    print(f"  ID={msg['id']}, Priority={msg['priority']}")

# 【Lambda用于过滤】filter()配合lambda筛选元素
# filter(条件函数, 可迭代对象) → 满足条件的元素
high_priority = list(filter(lambda x: x["priority"] >= 2, messages))
print(f"\n高优先级消息: {[m['id'] for m in high_priority]}")

# 【Lambda用于映射】map()配合lambda转换元素
# map(转换函数, 可迭代对象) → 转换后的新元素
# 注意:推导式[x["id"] for x in messages]更Pythonic
ids = list(map(lambda x: x["id"], messages))
print(f"所有ID: {ids}")

# 【LangGraph条件边示例】传统函数版本
def route_based_on_intent(state):
    """根据意图路由到不同节点"""
    # 字典映射:意图 → 节点名
    intent_routes = {
        "weather": "weather_node",
        "booking": "booking_node",
        "chat": "llm_node"
    }

    intent = state.get("intent", "chat")
    # 如果意图不在映射中,默认返回llm_node
    return intent_routes.get(intent, "llm_node")

# 【Lambda版本】一行实现同样逻辑
# 适合add_conditional_edges()的condition参数
route_lambda = lambda s: {
    "weather": "weather_node",
    "booking": "booking_node"
}.get(s.get("intent", "chat"), "llm_node")

# 测试
test_state = {"intent": "weather"}
print(f"\n路由结果: {route_based_on_intent(test_state)}")
print(f"Lambda 路由: {route_lambda(test_state)}")

预期结果:

按优先级排序:
  ID=1, Priority=1
  ID=3, Priority=2
  ID=2, Priority=3

高优先级消息: [3, 2]
所有ID: [3, 1, 2]

路由结果: weather_node
Lambda 路由: weather_node

6. 类型提示:为 LangGraph 做准备

🚀 30秒快速上手:最简类型提示

如果你是 Python 小白,先看这个最简单的类型提示:

python
# 类型提示 - 告诉别人(和自己)这个变量是什么类型
name: str = "小明"        # str 表示字符串(文字)
age: int = 18             # int 表示整数
score: float = 95.5       # float 表示小数

# 函数也可以加类型提示
def add(a: int, b: int) -> int:    # -> int 表示返回值是整数
    return a + b

result = add(5, 3)
print(result)             # 输出:8

💡 新手提示

  • 类型提示不是必须的,但能让代码更清晰
  • : 后面写类型,-> 后面写返回值类型
  • 常见类型:str(文字)、int(整数)、float(小数)、bool(真假)

类型提示(Type Hints)是 Python 3.5+ 引入的特性,在 LangGraph 中大量使用。

6.1 基础类型提示

python
# 代码示例:类型提示基础
from typing import List, Dict, Optional, Union

# 【变量类型提示】增强代码可读性和IDE智能提示
# 格式: 变量名: 类型 = 值
user_name: str = "Alice"
age: int = 25
score: float = 0.85
is_active: bool = True

# 【函数类型提示】声明输入输出类型,避免类型错误
# 格式: def 函数名(参数: 类型) -> 返回类型:
def format_message(role: str, content: str) -> dict:
    """格式化消息

    Args:
        role: 角色(user/assistant)
        content: 消息内容

    Returns:
        格式化的消息字典
    """
    return {"role": role, "content": content}

# 【复合类型】List[元素类型]表示列表中元素的类型
# IDE可以自动提示messages[0]有哪些字典键
messages: List[dict] = [
    {"role": "user", "content": "hello"},
    {"role": "assistant", "content": "hi"}
]

# 【Dict类型】Dict[键类型, 值类型]
# any表示任意类型(不推荐,应尽量具体化)
state: Dict[str, any] = {
    "user_id": "123",
    "count": 5
}

# 【Optional类型】等价于Union[指定类型, None]
# 明确告诉调用方:此函数可能返回None,需要处理空值情况
def get_user_name(user_id: str) -> Optional[str]:
    """获取用户名,可能返回 None"""
    users = {"123": "Alice", "456": "Bob"}
    return users.get(user_id)  # dict.get()可能返回None

result = get_user_name("789")
print(f"查询结果: {result}")  # None

# 【Union类型】表示"多选一",函数可以接受多种类型
# 常用于处理API参数(可能是字符串或数字)
def process_input(value: Union[str, int, float]) -> str:
    """处理不同类型的输入"""
    return f"处理: {value} (类型: {type(value).__name__})"

print(process_input("hello"))
print(process_input(123))
print(process_input(3.14))

预期结果:

查询结果: None
处理: hello (类型: str)
处理: 123 (类型: int)
处理: 3.14 (类型: float)

6.2 LangGraph 风格的类型提示

python
# 代码示例:模拟 LangGraph 的类型定义
from typing import TypedDict, List, Literal

# 【TypedDict】定义严格的字典结构(类似TypeScript的Interface)
# 比普通dict更安全:IDE可检查字段拼写错误,mypy可做静态类型检查
class Message(TypedDict):
    # Literal限定role只能是这三个值之一,避免拼写错误
    role: Literal["user", "assistant", "system"]
    content: str

# 【嵌套TypedDict】LangGraph状态的标准定义方式
# 定义清晰的状态结构,便于团队协作和维护
class AgentState(TypedDict):
    messages: List[Message]  # 消息列表,每个元素都是Message类型
    intent: str
    next_node: str

# 【类型化函数】明确返回AgentState,IDE可以智能提示字段
def create_initial_state(user_message: str) -> AgentState:
    """创建初始状态

    【设计模式】工厂函数,统一状态初始化逻辑
    避免在各处重复构建状态结构
    """
    return {
        "messages": [
            {"role": "user", "content": user_message}
        ],
        "intent": "unknown",
        "next_node": "classifier"
    }

# 【节点函数类型签名】LangGraph标准模式
# 输入AgentState → 处理 → 输出AgentState
def classify_intent(state: AgentState) -> AgentState:
    """意图分类节点"""
    user_message = state["messages"][-1]["content"]

    # 【状态更新】TypedDict仍然是可变字典,可以直接修改
    if "天气" in user_message:
        state["intent"] = "weather"
        state["next_node"] = "weather_tool"
    else:
        state["intent"] = "general"
        state["next_node"] = "llm"

    # 【类型安全】返回类型必须匹配AgentState,否则mypy报错
    return state

# 测试
initial = create_initial_state("今天天气怎么样?")
print("初始状态:")
print(initial)

updated = classify_intent(initial)
print("\n分类后:")
print(f"意图: {updated['intent']}")
print(f"下一节点: {updated['next_node']}")

预期结果:

初始状态:
{'messages': [{'role': 'user', 'content': '今天天气怎么样?'}], 'intent': 'unknown', 'next_node': 'classifier'}

分类后:
意图: weather
下一节点: weather_tool

7. 实战案例:构建简单的意图分类器

让我们综合运用本章所有知识,构建一个完整的意图分类系统。

python
# 代码示例:完整的意图分类器
from typing import Dict, List, Tuple

class IntentClassifier:
    """简单的意图分类器

    【架构设计】
    - 基于规则的分类器(适合明确意图场景)
    - 实际生产应使用BERT分类模型或Few-Shot LLM
    - 演示OOP、类型提示、数据处理的综合应用
    """

    def __init__(self):
        """初始化分类器

        【构造函数】创建实例时自动调用,初始化实例属性
        self.xxx 是实例属性,每个对象独立拥有
        """
        # 【意图模式库】字典存储意图与关键词的映射
        # 实际应用可从配置文件加载,支持动态更新
        self.intent_patterns = {
            "weather": ["天气", "温度", "下雨", "晴天"],
            "booking": ["预订", "订票", "预约", "安排"],
            "calculator": ["计算", "加", "减", "乘", "除", "="],
            "greeting": ["你好", "hi", "hello", "早上好"]
        }
        # 【历史记录】List[Dict]存储所有分类结果,用于统计分析
        # 生产中应限制大小或持久化到数据库
        self.history: List[Dict] = []

    def classify(self, message: str) -> Dict[str, any]:
        """分类消息

        【核心算法】关键词匹配 + 得分计算
        时间复杂度:O(意图数 × 关键词数),适合小规模场景

        Args:
            message: 用户消息

        Returns:
            分类结果字典,包含意图、置信度、详细得分
        """
        # 【预处理】统一转小写,提高匹配率
        message_lower = message.lower()
        scores = {}

        # 【得分计算】遍历所有意图,统计关键词命中次数
        for intent, keywords in self.intent_patterns.items():
            # 列表推导式 + sum:简洁统计匹配数
            # sum(1 for ...) 等价于 len([kw for kw in keywords if ...])
            score = sum(1 for kw in keywords if kw in message_lower)
            if score > 0:
                scores[intent] = score

        # 【意图判定】找到得分最高的意图
        if scores:
            # max()的key参数:按元组第二项(分数)排序
            # best_intent是(intent, score)元组
            best_intent = max(scores.items(), key=lambda x: x[1])
            intent, confidence = best_intent[0], best_intent[1] / 5
        else:
            # 【未匹配处理】返回unknown,置信度0
            intent, confidence = "unknown", 0.0

        # 【结果封装】标准化输出格式
        result = {
            "message": message,
            "intent": intent,
            "confidence": min(confidence, 1.0),  # 限制最大值为1.0
            "all_scores": scores  # 保留所有得分,便于调试
        }
        # 【历史记录】追加到实例属性,支持后续分析
        self.history.append(result)

        return result

    def batch_classify(self, messages: List[str]) -> List[Dict]:
        """批量分类

        【批处理优化】列表推导式简洁实现
        实际应用可并行处理提升性能
        """
        return [self.classify(msg) for msg in messages]

    def get_stats(self) -> Dict[str, int]:
        """获取统计信息

        【数据分析】统计各意图出现次数
        用于分析用户行为模式、优化意图库
        """
        intent_counts = {}
        for record in self.history:
            intent = record["intent"]
            # dict.get()配合默认值:简洁实现计数累加
            intent_counts[intent] = intent_counts.get(intent, 0) + 1
        return intent_counts

    def clear_history(self):
        """清空历史

        【内存管理】防止history无限增长导致内存溢出
        """
        self.history = []


# ===== 使用示例 =====

# 1. 创建分类器
classifier = IntentClassifier()

# 2. 单条消息分类
test_messages = [
    "今天北京天气怎么样?",
    "帮我预订明天的机票",
    "计算 123 + 456",
    "你好,请问有什么可以帮助的吗?",
    "这是一个随机消息"
]

print("=== 单条分类 ===")
for msg in test_messages:
    result = classifier.classify(msg)
    print(f"\n消息: {msg}")
    print(f"  意图: {result['intent']}")
    print(f"  置信度: {result['confidence']:.0%}")
    if result['all_scores']:
        print(f"  得分详情: {result['all_scores']}")

# 3. 批量分类
batch_messages = [
    "明天会下雨吗?",
    "订一张去上海的票",
    "早上好!"
]

print("\n\n=== 批量分类 ===")
batch_results = classifier.batch_classify(batch_messages)
for result in batch_results:
    print(f"{result['message']}{result['intent']}")

# 4. 统计分析
print("\n\n=== 统计信息 ===")
stats = classifier.get_stats()
for intent, count in sorted(stats.items(), key=lambda x: x[1], reverse=True):
    print(f"  {intent}: {count} 条")

print(f"\n总处理消息数: {len(classifier.history)}")

预期结果:

=== 单条分类 ===

消息: 今天北京天气怎么样?
  意图: weather
  置信度: 20%
  得分详情: {'weather': 1}

消息: 帮我预订明天的机票
  意图: booking
  置信度: 20%
  得分详情: {'booking': 1}

消息: 计算 123 + 456
  意图: calculator
  置信度: 20%
  得分详情: {'calculator': 1}

消息: 你好,请问有什么可以帮助的吗?
  意图: greeting
  置信度: 20%
  得分详情: {'greeting': 1}

消息: 这是一个随机消息
  意图: unknown
  置信度: 0%


=== 批量分类 ===
明天会下雨吗? → weather
订一张去上海的票 → booking
早上好! → greeting


=== 统计信息 ===
  weather: 2 条
  booking: 2 条
  greeting: 2 条
  calculator: 1 条
  unknown: 1 条

总处理消息数: 8

8. 本章总结

✅ 你已经掌握

  1. 变量与数据类型

    • 基础类型:str, int, float, bool, None
    • 字符串操作:strip(), lower(), split(), f-string
  2. 数据结构

    • List:对话历史、消息队列
    • Dict:Agent 状态、配置
    • Tuple:固定格式、多返回值
    • Set:去重、集合运算
  3. 控制流

    • if-elif-else:意图路由
    • for 循环:批量处理
    • while 循环:重试机制
    • 列表/字典推导式
  4. 函数

    • 函数定义与调用
    • 参数(默认、可选、**kwargs)
    • Lambda 表达式
    • 多返回值
  5. 类型提示

    • 基础类型:str, int, List, Dict
    • 高级类型:Optional, Union, TypedDict
    • 函数注解

📚 下一步

准备好了吗?接下来我们将学习:

  • 0.2-面向对象与工程实践.md - 构建生产级 Agent
  • 0.3-AI开发工具链.md - 掌握 AI 生态系统

💡 练习建议

  1. 修改意图分类器,添加更多意图类别
  2. 实现一个简单的对话管理器
  3. 尝试使用类型提示重构你的代码

🎯 记住:Python 的核心不是语法,而是用它解决实际问题。在 LangGraph 开发中,你会不断使用这些基础知识来构建强大的 AI Agent!

基于 MIT 许可证发布。内容版权归作者所有。